package com.shujia.spark.optimize

import java.sql.DriverManager

import org.apache.spark.{SparkConf, SparkContext}

object Demo0ForeachPartition {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("app")
    val sc = new SparkContext(conf)

    val student = sc.textFile("spark/data/students.txt")

    val reStudent = student.repartition(4)

    /**
      * 将rdd的数据保存到外部数据库 mysql
      *
      * foreach  ： 每一行数据都需要创建一个jdbc连接，   task内部不能使用Driver端的网络连接对象，  网络连接不能再网络中传输   ，网络连接不能序列化
      *
      * foreachPartition : 每一个分区创建一个连接，减少了连接的数量，提高了效率
      */


    /*    reStudent.foreach(line => {
          //创建mysql连接
          Class.forName("com.mysql.jdbc.Driver")
          val con = DriverManager.getConnection("jdbc:mysql://node1:3306/test?characterEncoding=utf-8", "test", "123456")

          val stat = con.prepareStatement("insert into student(value) values(?)")

          stat.setString(1, line)

          stat.execute()

          con.close()

        })*/


    reStudent.foreachPartition(iter => {
      //创建mysql连接
      Class.forName("com.mysql.jdbc.Driver")
      val con = DriverManager.getConnection("jdbc:mysql://node1:3306/test?characterEncoding=utf-8", "test", "123456")

      iter.foreach(line => {
        val stat = con.prepareStatement("insert into student(value) values(?)")
        stat.setString(1, line)
        stat.execute()
      })
      con.close()
    })


  }
}
